大家好！歡迎來到鐵人賽第九天。

昨天，我們成功透過 Webhook，讓 Wazuh Server 在觸發告警時，主動對 FastAPI 接收站發送 HTTP POST 請求。

但仔細想想，目前的程式其實只做了一件事：確認資料有成功送達，卻還不知道告警的內容到底是什麼。

Wazuh 傳遞過來的資料是一大包 JSON 格式的 Payload，裡面包含了事件時間、主機名稱、觸發規則，甚至還可能包含 CVE 漏洞資訊。

如果我們想要讓後續的自動化系統根據告警內容進行判斷，就不能只把整包資料印出來，而是必須從中找出真正有用的資訊。

因此，今天我們要實作「資料解析（Data Parsing）」，利用 Python 將原始 JSON 拆解成有意義的資安情報，讓 FastAPI 從單純的資料接收站，逐步進化成能夠理解告警內容的後端服務。

一、先認識 Wazuh Alert JSON：告警資料到底長什麼樣子？

在開始寫程式之前，我們必須先了解 Wazuh 傳送過來的 JSON 結構。

JSON 是一種常見的資料交換格式，使用鍵值對的方式儲存資訊。

例如：

{
  "name": "Wendy",
  "age": 20
}

這裡的 name 和 age 就是欄位名稱，而後面的值則是對應的資料。

不過，Wazuh 的告警資料通常比這個例子複雜許多，因為它會將不同類型的資訊分別放在不同的物件中，形成一層包一層的巢狀結構（Nested Dictionary）。

以下是簡化後的告警資料範例：

{
  "timestamp": "2026-09-28T16:30:00.000+0800",
  "rule": {
    "level": 7,
    "description": "Vulnerability severity: High",
    "id": "23505"
  },
  "agent": {
    "id": "001",
    "name": "LAPTOP-3007V8UH",
    "ip": "192.168.1.100"
  },
  "data": {
    "vulnerability": {
      "cve": "CVE-2023-12345",
      "severity": "High",
      "package": {
        "name": "python3",
        "version": "3.10.6"
      }
    }
  }
}

這裡的資料是示意範例，實際欄位會依照 Wazuh 的告警類型而有所不同。

從這份資料中，我們可以看到幾個重要區塊：

1. rule：告警規則資訊

"rule": {
  "level": 7,
  "description": "Vulnerability severity: High",
  "id": "23505"
}

這個區塊主要包含觸發告警的規則資訊。

• level：告警等級。
• description：告警描述。
• id：觸發的規則編號。

透過這些欄位，我們可以快速了解系統為什麼產生告警，以及告警的等級。

2. agent：被監控的主機資訊

"agent": {
  "id": "001",
  "name": "LAPTOP-3007V8UH",
  "ip": "192.168.1.100"
}

這個區塊描述產生告警的 Agent，也就是被監控的主機。

其中包含主機名稱、Agent ID 與 IP 位址，能幫助我們辨識是哪一台主機發生了事件。

3. data.vulnerability：漏洞相關資訊

"data": {
  "vulnerability": {
    "cve": "CVE-2023-12345",
    "severity": "High",
    "package": {
      "name": "python3",
      "version": "3.10.6"
    }
  }
}

這個區塊則包含漏洞偵測相關的資料，例如：

• CVE 編號。
• 漏洞嚴重程度。
• 受影響套件名稱。
• 套件版本。

不過，並不是所有告警都會包含 vulnerability 這個欄位。

例如，登入失敗事件可能只有登入帳號、來源 IP 與事件描述，並不會有 CVE 資訊。

因此，我們在解析資料時，必須考慮不同類型的告警可能具有不同的欄位結構。

二、改寫 FastAPI 路由：讓程式真正讀懂告警

了解 JSON 結構後，接下來就要回到 main.py，將原本只會印出「收到資料」的路由重新改寫。

在 Python 中，如果想要取得巢狀字典的資料，可以直接使用中括號：

level = payload["rule"]["level"]

這樣就能取得告警等級。

但如果 rule 或 level 不存在，程式就可能拋出 KeyError。

而當我們處理不同類型的資安告警時，欄位缺漏是很常見的情況。

因此，我們需要讓程式具備更好的資料驗證與錯誤處理能力。

這次，我們會使用兩個重要工具：

Pydantic： 定義資料模型，驗證收到的資料是否符合預期。

.get() 方法： 在存取可能不存在的字典欄位時，提供預設值，避免因為缺少某個鍵值而直接發生錯誤。

三、導入 Pydantic：建立告警資料模型

Pydantic 是 Python 常用的資料驗證與解析套件，而 FastAPI 會利用它來處理請求資料。

我們可以透過建立資料模型，明確定義哪些欄位需要存在、欄位應該是什麼型別，以及哪些資料可以省略。

例如：

from pydantic import BaseModel, Field
from typing import Optional, Dict, Any

class RuleModel(BaseModel):
    level: int = Field(default=0, description="告警等級")
    description: str = Field(
        default="No description",
        description="告警描述"
    )

class AgentModel(BaseModel):
    name: str = Field(default="Unknown", description="主機名稱")
    ip: Optional[str] = Field(default=None, description="主機 IP")

class WazuhPayload(BaseModel):
    rule: RuleModel
    agent: AgentModel
    data: Optional[Dict[str, Any]] = None

這段程式碼主要做了三件事。

1. 定義 RuleModel

class RuleModel(BaseModel):
    level: int = Field(default=0)
    description: str = Field(default="No description")

這個模型用來描述告警規則。

我們將 level 定義為整數，並將 description 定義為字串。

如果這兩個欄位沒有提供，就會使用預設值。

2. 定義 AgentModel

class AgentModel(BaseModel):
    name: str = Field(default="Unknown")
    ip: Optional[str] = None

這個模型用來描述被監控的主機。

其中，Optional[str] 表示這個欄位可以是字串，也可以是 None。

這讓我們在處理缺少 IP 位址的情況時，更有彈性。

3. 定義 WazuhPayload

class WazuhPayload(BaseModel):
    rule: RuleModel
    agent: AgentModel
    data: Optional[Dict[str, Any]] = None

這個模型是整份告警資料的入口。

rule 與 agent 分別使用前面定義的資料模型，而 data 則允許接收不同類型的資料。

這樣一來，即使不同告警的 data 結構不完全相同，也能先將它保留，交由後續程式判斷。

需要注意的是，這份模型只涵蓋我們目前需要的欄位，並不是完整的 Wazuh Alert Schema。

四、實作資料解析：從 JSON 萃取關鍵情報

完成資料模型後，我們就可以開始改寫 FastAPI 路由。

這次除了接收資料之外，還會加入日誌系統，讓終端機能夠清楚顯示告警內容。

1. 為什麼使用 logging，而不是 print？

在初期測試時，使用 print() 輸出訊息確實很方便。

但當系統逐漸複雜，我們就需要更有組織的日誌紀錄方式。

Python 的 logging 模組可以讓我們設定日誌等級、時間格式與訊息內容。

例如：

• INFO：一般資訊。
• WARNING：需要注意的事件。
• ERROR：發生錯誤的情況。

透過這種方式，我們就能更容易區分一般告警與需要進一步處理的事件。

2. 完整程式碼

以下是整合資料模型、日誌輸出與漏洞資訊解析後的 main.py：

from fastapi import FastAPI
from pydantic import BaseModel, Field
from typing import Optional, Dict, Any
import logging
import uvicorn

# 初始化日誌系統
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)-7s | %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S"
)

logger = logging.getLogger("Wazuh-Webhook")


# 定義資料模型
class RuleModel(BaseModel):
    level: int = Field(default=0, description="告警等級")
    description: str = Field(
        default="No description",
        description="告警描述"
    )


class AgentModel(BaseModel):
    name: str = Field(default="Unknown", description="主機名稱")
    ip: Optional[str] = Field(default=None, description="主機 IP")


class WazuhPayload(BaseModel):
    rule: RuleModel
    agent: AgentModel
    data: Optional[Dict[str, Any]] = None


# 建立 FastAPI
app = FastAPI(
    title="SecOps Backend API",
    version="1.0.0"
)


@app.post("/webhook/wazuh")
async def receive_wazuh_alert(payload: WazuhPayload):

    # 取得基本告警資訊
    logger.info(
        f"[告警接收] 主機: {payload.agent.name} | "
        f"等級: {payload.rule.level} | "
        f"描述: {payload.rule.description}"
    )

    # 取得漏洞相關資料
    if payload.data and "vulnerability" in payload.data:

        vuln = payload.data["vulnerability"]

        cve_id = vuln.get("cve", "Unknown CVE")
        severity = vuln.get("severity", "Unknown")

        package = vuln.get("package") or {}
        pkg_name = package.get("name", "Unknown Package")

        logger.warning(
            f"[漏洞情報] {cve_id} | "
            f"受影響套件: {pkg_name} | "
            f"嚴重程度: {severity}"
        )

    return {
        "status": "success",
        "message": "Alert processed"
    }


if __name__ == "__main__":
    uvicorn.run(
        "main:app",
        host="0.0.0.0",
        port=8000,
        reload=True
    )

3. .get() 為什麼這麼重要？

在程式中，我們使用了：

cve_id = vuln.get("cve", "Unknown CVE")

這行程式碼的意思是：

如果 vuln 裡面有 cve 這個欄位，就取得它的值；如果沒有，就回傳 "Unknown CVE"。

這樣就能避免單純因為缺少某個欄位而發生 KeyError。

另外，這裡也使用了：

package = vuln.get("package") or {}

這是為了避免 package 欄位不存在或是 None 時，後續呼叫 .get() 產生錯誤。

這些寫法可以讓程式在面對不完整資料時更加穩健。

不過，如果資料結構本身不符合預期，仍然需要進一步的錯誤處理，不能只靠 .get() 解決所有問題。

五、驗證解析結果：再次觸發登入失敗告警

完成程式修改並重新啟動 FastAPI 後，我們再次回到 Ubuntu 測試主機。

這次，我們使用：

su - fakeuser

嘗試切換到不存在的使用者，並輸入錯誤密碼，讓系統產生登入失敗事件。

當 Wazuh 偵測到符合條件的事件後，就會透過 Webhook 將告警傳送給 FastAPI。

接著，我們切回 Windows 的終端機，觀察日誌輸出。

這次，終端機不再只有單調的 200 OK，而是出現了整理過後的告警資訊：

2026-09-28 16:30:00 | INFO    | [告警接收] 主機: vm1 | 等級: 5 | 描述: User missed the password to change UID (user id).

從這行輸出中，我們可以快速辨識：

• 時間： 告警被後端記錄的時間。
• 主機： 產生告警的主機名稱。
• 等級： Wazuh 告警等級。
• 描述： 觸發的事件內容。

這代表我們已經成功將原本的 JSON 資料轉換成更容易閱讀的日誌。

而且，這次測試的是登入失敗事件，不包含漏洞資訊，因此程式也沒有強行尋找 CVE 欄位。

這正是我們希望達成的效果：讓系統能夠依照不同的告警類型，選擇性地解析需要的資料。

六、今天的成果總結

今天，我們成功將 FastAPI 從單純的告警接收站，進一步改造成具備基本資料解析能力的後端服務。

回顧今天完成的內容：

• 了解 Wazuh Alert JSON 的巢狀資料結構。
• 認識 rule、agent 與 data.vulnerability 的用途。
• 使用 Pydantic 建立資料模型。
• 利用 .get() 處理可能缺少的欄位。
• 導入 Python logging 模組，讓告警輸出更加清楚。
• 成功解析登入失敗告警，並驗證程式能處理不含漏洞資訊的事件。

從昨天的：

Wazuh 偵測告警 → Webhook 傳送 → FastAPI 接收

到今天的：

Wazuh 偵測告警 → Webhook 傳送 → FastAPI 接收 → JSON 解析 → 萃取關鍵情報

我們已經讓系統跨出了重要的一步。

不過，目前這些情報仍然只停留在 Python 的日誌輸出中，還沒有進一步整合到自動化通報流程。

七、明天預告：導入 n8n，讓資安情報自動推播

今天，我們已經成功讓 FastAPI 讀懂 Wazuh 傳來的告警資料。

但如果每次發生事件，都需要人工盯著終端機才能知道結果，那麼自動化的價值仍然有限。

因此，明天我們將正式導入開源自動化工具 n8n，嘗試將 FastAPI 整理好的資安情報串接到後續的通知流程。

我們希望未來當系統偵測到異常登入或漏洞告警時，就能自動整理重要資訊，並透過通訊軟體將通知傳送給管理者。

從今天的「讀懂告警」，走向明天的「自動通知」。

我們明天見！